package mapred.joinmode;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.MultipleInputs;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

import java.io.IOException;

public class JoinRunner {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Configuration conf = new Configuration();
        String[] otherArgs = null;
        otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();

        if (otherArgs.length != 4) {
            // 连接模式，需要用到必须2个输入文件夹，作为2个表的数据来源
            // 第三个参数用来接收连接的类型：inner leftouter rightouter
            // 最后一个参数才是输出文件夹的路径
            System.err.println("Usage: mapred.joinmode.JoinRunner <UserPath> <CommentPath> <jointype> <outputPath>");
            System.exit(2);
        }

        Job job = null;
        job = Job.getInstance();

        job.setJobName("Join");
        job.setJarByClass(JoinRunner.class);

        // 通过MultipleInputs类的静态方法addInputPath，为当前Job添加多个输入源
        // 接收的数据，具体使用哪种输入格式解析，由addInputPath函数的第三个参数来指定
        // MultipleInputs实现了在一个Job中，可以同时使用多个Mapper来出来文件
        // 每个Mapper可以对应不同的输入格式
        // 使用了MultipleInputs之后，就不需要再使用job.setMapperClass来单独设置Mapper的类了
        MultipleInputs.addInputPath(job, // 当前Job的实例
                new Path(otherArgs[0]), // 命令行参数的一个路径，作为User的数据
                TextInputFormat.class, // 指定使用何种格式解析数据，传递给Mapper
                UserJoinMapper.class   // 指定当前文件（user)使用哪个Mapper来处理
        );

        MultipleInputs.addInputPath(job,
                new Path(otherArgs[1]),
                TextInputFormat.class,
                CommentJoinMapper.class
        );

        // 将从命令行接收到的连接类型通过配置参数传递到Reduce中
        job.getConfiguration().set("join.type", otherArgs[2]);

        // 设置Reducer类，真正的Join连接过程就是在这个类中实现的。
        job.setReducerClass(UserJoinReducer.class);

        // 设置Mapper的输出类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);

        // 设置Reducer的输出类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        // 设置输出文件夹
        FileOutputFormat.setOutputPath(job, new Path(otherArgs[3]));

        // 启用对输出结果文件的压缩功能
        FileOutputFormat.setCompressOutput(job, true);
        // 对输出文件使用BZip2进行压缩
        FileOutputFormat.setOutputCompressorClass(job, org.apache.hadoop.io.compress.BZip2Codec.class);

        FileSystem fs = FileSystem.get(conf);
        if (fs.exists(new Path(otherArgs[3]))) {
            fs.delete(new Path(otherArgs[3]), true);
        }

        System.exit(job.waitForCompletion(true) ? 0 : 1);

    }
}
